package com.xf.day07_graphx
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
/**
 * 操作图结构的变换
 *     (1,1)
 *     (3,2)
 *     (4,1)
 *     (5,1)
 *     (6,2)
 *
 */
object OperatorGraphStructure {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("testGraphx2").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")

    // 创建一个顶点集的 RDD,  VertexId 一个 Long 类型, 顶点属性是一个 二元组
    val users : RDD[( VertexId, (String, Int))] = sc.parallelize(
      Array(
        ( 1L, ("菲菲", 22)),
        ( 2L, ("乔乔", 17)),
        ( 3L, ("王强", 19)),
        ( 4L, ("王刚", 21)),
        ( 5L, ("李倩", 20)),
        ( 6L, ("林峰", 25))
      ))

    // 创建一个边集的 RDD
    val relationships :RDD[ Edge[Int]] = sc.parallelize(
      Array( Edge(1L ,3L, 15) ,
        Edge(2L ,1L, 10) ,
        Edge(2L ,3L, 21) ,
        Edge(2L ,4L, 22) ,
        Edge(3L ,6L, 10) ,
        Edge(4L ,5L, 17) ,
        Edge(5L ,6L, 20)
      ))

    // 定义边中 用户缺失时的 默认 (缺失) 用户
    val defaultUser = ("某某", 18)

    // 使用 users 和 relationships 两个RDD 实例化 Graph类建立一个 Graph对象
    val userGraph = Graph( users, relationships, defaultUser)

    // 翻转图
    // reverseGraph(userGraph)

    // 获取 子图
    // getSubGraph(userGraph)

    // 获取子图 2
    getGraph2(userGraph)

  }

  private def getGraph2(userGraph: Graph[(String, Int), Int]): Unit = {
    /**
     * 注意:
     * 具体的结果为:
     * Edge(1,3,15)
     * Edge(2,3,21)
     */
    // 获取边的属性 attr < 22 的图
    // userGraph.subgraph( epred = edge => edge.attr < 22 )
    val userGraph2: Graph[(String, Int), Int] = userGraph.subgraph(epred = edge => edge.dstAttr._1 == "王强")
    userGraph2.edges.collect().foreach(println)
    println("===========================> ")

    /**
     * 获取 源点年龄 大于 目标年龄 的子图
     *
     * 具体的结果为：
     * Edge(1,3,15)
     * Edge(4,5,17)
     */
    val userGraph3 = userGraph.subgraph(epred = edge => edge.srcAttr._2 > edge.dstAttr._2)
    userGraph3.edges.collect().foreach(println)
    println("===========================> ")

    /**
     * 对 顶点 和边同时操作 获取子图
     * userGraph.subgraph( epred = edge => edge.attr < 22 ) 执行的结果如下:
     * Edge(1,3,15)
     * Edge(2,1,10)
     * Edge(2,3,21)
     * Edge(3,6,10)
     * Edge(4,5,17)
     * Edge(5,6,20)
     *
     *
     * userGraph.subgraph( epred = edge => edge.attr < 22 , vpred = (id, attr) => attr._2 < 22 ) 执行的结果为:
     * Edge(2,3,21)
     * Edge(4,5,17)
     */
    val userGraph4: Graph[(String, Int), Int] = userGraph.subgraph(epred = edge => edge.attr < 22, vpred = (id, attr) => attr._2 < 22)
    userGraph4.edges.collect().foreach(println)
    println(" ==============================> ")

    /**
     * 输出子图的顶点 的结果为:
     * (2,(乔乔,17))
     * (3,(王强,19))
     * (4,(王刚,21))
     * (5,(李倩,20))
     */
    userGraph4.vertices.collect().foreach(println)
  }

  /**
   * 注意:
   *    1.获取子图的方法
   */
  private def getSubGraph(userGraph: Graph[(String, Int), Int]): Unit = {
    // 翻转图
    // reverseGraph(userGraph)

    // 获取顶点的 age < 22 的子图
    // 获取顶点的 age < 21 的子图    ========> 获取的是 一个断点图
    val subGraph1 = userGraph.subgraph( vpred = (id, attr) => attr._2 < 22 )

    /**
     * 输出子图顶点
     * (2,(乔乔,17))
     * (3,(王强,19))
     * (4,(王刚,21))
     * (5,(李倩,20))
     */
    subGraph1.vertices.collect().foreach(println)
    println("===========================>")

    /**
     * 输出子图的边
     * Edge(2,3,21)
     * Edge(2,4,22)
     * Edge(4,5,17)
     */
    subGraph1.edges.collect().foreach(println)
  }

  /**
   *   翻转图
   */
  private def reverseGraph(userGraph: Graph[(String, Int), Int]): Unit = {
    // 输出所有顶点的入度
    userGraph.inDegrees.collect().foreach(println)
    println("=================================>")

    // 有向边 翻转  ==> 得到新图
    val reverseGraph = userGraph.reverse

    /**
     * 翻转后的图  ==> 直接打印出 入度
     */
    reverseGraph.inDegrees.collect().foreach(println)
  }



}
